#include "producer/client/RouteInfoDownloader.h"
#include "producer/client/ProducerMediator.h"
#include "codec/ProtocolCodec.h"

namespace zeroplus::mq::producer
{
    RouteInfoDownloader::RouteInfoDownloader(std::shared_ptr<ProducerMediator> mediator, std::string uri)
        : mediator_(mediator)
    {
        client_ = std::make_unique<ClientNode>(uri);
    }

    RouteInfoDownloader::~RouteInfoDownloader()
    {
    }

    void RouteInfoDownloader::Start()
    {
        loop_.set_name("RouteInfoDownloader");
        loop_.async_run();
        timer_ = std::make_shared<Timer>(&loop_,
                                         NetworkFrequency::queryRouteInfoFrequency,
                                         Timer::kInfinite,
                                         std::bind(&RouteInfoDownloader::SendQueryRouteInfoRequest, this));
        timer_->start();
    }

    void RouteInfoDownloader::SendQueryRouteInfoRequest()
    {
        std::string pendingTopic;
        if (!pendingTopicQueue_.try_dequeue(pendingTopic))
        {
            return;
        }
        transport::QueryRouteInfoRequest request;
        request.set_topic(pendingTopic);
        transport::RequestCommand requestCmd;
        requestCmd.set_code(transport::RequestCode::QUERY_ROUTEINFO_BY_TOPIC);
        requestCmd.set_body(ProtocolCodec<transport::QueryRouteInfoRequest>::Encode(request).to_string());
        auto requestBytes = ProtocolCodec<transport::RequestCommand>::Encode(requestCmd);
        LOG_DEBUG("发送路由查询请求");
        auto [responseBytes, res] = client_->InvokeSync(requestBytes);
        auto replyCmd = ProtocolCodec<transport::ReplyCommand>::Decode(responseBytes);
        mediator_->Notify(DownLoader, Bytes::from_string(replyCmd.body()));
    }

    void RouteInfoDownloader::InternalReceive(const std::string &topic)
    {
        pendingTopicQueue_.enqueue(topic);
    }
}
